package cn.doitedu.rtdw.etl;

import cn.doitedu.rtdw.udfs.TimeStampTruncateUdf;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_流量olap支撑轻度聚合 {

    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中dwd层的行为日志明细
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,device_type string                                "+
                        "     ,release_channel string                                "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );



        // 创建doris目标表的映射表
        tenv.executeSql("CREATE TABLE traffic_olap_base_01_doris(\n" +
                "    dt                 date\n" +
                "\t,time_60m          varchar\n" +
                "    ,time_30m          varchar\n" +
                "    ,time_10m          varchar\n" +
                "    ,time_5m           varchar\n" +
                "    ,time_1m           varchar\n" +
                "\t,province          varchar\n" +
                "\t,city              varchar\n" +
                "\t,region            varchar\n" +
                "\t,device_type       varchar\n" +
                "\t,release_channel   varchar\n" +
                "\t,page_type         varchar\n" +
                "\t,page_service      varchar\n" +
                "\t,page_url          varchar\n" +
                "\t,is_new            boolean\n" +
                "\t,user_id           bigint \n" +
                "\t,session_id        string\n" +
                "\t,pv                bigint  \n" +
                ")  with (                                          \n" +
                "    'connector' = 'doris',                         \n" +
                "    'fenodes' = 'doitedu:8030',                    \n" +
                "    'table.identifier' = 'dws.traffic_olap_base_01', \n" +
                "    'username' = 'root',                           \n" +
                "    'password' = 'root',                           \n" +
                "    'sink.label-prefix' = 'doit42_no3'  ,\n" +
                "    'sink.properties.columns' = 'dt,time_60m,time_30m,time_10m,time_5m,time_1m,province,city,region,device_type,release_channel,page_type,page_service,page_url,is_new,user_id,session_id,pv,uv=to_bitmap(user_id),ses_cnt=hll_hash(session_id)' \n" +
                ")");





        tenv.createTemporaryFunction("time_trunc", TimeStampTruncateUdf.class);
        tenv.executeSql("insert into traffic_olap_base_01_doris " +
                "with tmp as (\n" +
                "    select\n" +
                "       *, \n" +
                "\t   date_format(create_time,'yyyy-MM-dd')=current_date as is_new\n" +
                "    from dwd_events_kafka\n" +
                "    where event_id = 'page_load'\n" +
                ")\n" +
                " " +
                "SELECT\n" +
                "\tto_date(date_format(window_start,'yyyy-MM-dd')) as dt,\n" +
                "\ttime_trunc(window_start,60) as time_60m, \n" +
                "\ttime_trunc(window_start,30) as time_30m,\n" +
                "\ttime_trunc(window_start,10) as time_10m,\n" +
                "\ttime_trunc(window_start,5)  as time_5m,  \n" +
                "\ttime_trunc(window_start,1)  as time_1m,\n" +
                "\tprovince,\n" +
                "\tcity,\n" +
                "\tregion,\n" +
                "\tdevice_type,\n" +
                "\trelease_channel,\n" +
                "\tpage_type,\n" +
                "\tpage_service,\n" +
                "\tproperties['url'] as page_url,\n" +
                "\tis_new,\n" +
                "\tuser_id,\n" +
                "\tsession_id,\n" +
                "\tcount(1) as pv\n" +
                "FROM TABLE(\n" +
                "    TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE )\n" +
                ")\n" +
                "GROUP BY \n" +
                "    window_start,\n" +
                "\twindow_end,\n" +
                "\tprovince,\n" +
                "\tcity,\n" +
                "\tregion,\n" +
                "\tdevice_type,\n" +
                "\trelease_channel,\n" +
                "\tpage_type,\n" +
                "\tpage_service,\n" +
                "\tproperties['url'],\n" +
                "\tis_new,\n" +
                "\tuser_id,\n" +
                "\tsession_id");

    }


}
